/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package org.apache.coyote;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import javax.servlet.RequestDispatcher;

import org.apache.tomcat.util.ExceptionUtils;
import org.apache.tomcat.util.buf.ByteChunk;
import org.apache.tomcat.util.buf.MessageBytes;
import org.apache.tomcat.util.http.parser.Host;
import org.apache.tomcat.util.log.UserDataHelper;
import org.apache.tomcat.util.net.AbstractEndpoint;
import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState;
import org.apache.tomcat.util.net.DispatchType;
import org.apache.tomcat.util.net.SSLSupport;
import org.apache.tomcat.util.net.SocketEvent;
import org.apache.tomcat.util.net.SocketWrapperBase;
import org.apache.tomcat.util.res.StringManager;

/**
 * Provides functionality and attributes common to all supported protocols
 * (currently HTTP and AJP) for processing a single request/response.
 */
public abstract class AbstractProcessor extends AbstractProcessorLight implements ActionHook {

    private static final StringManager sm = StringManager.getManager(AbstractProcessor.class);

    // Used to avoid useless B2C conversion on the host name.
    protected char[] hostNameC = new char[0];//保存主机名。

    protected Adapter adapter;//coyote适配器，用于将请求代理到Coyote processor执行器中执行。
    protected final AsyncStateMachine asyncStateMachine;//异步处理状态机。
    private volatile long asyncTimeout = -1;//异步超时时间。
    /*
     * Tracks the current async generation when a timeout is dispatched. In the
     * time it takes for a container thread to be allocated and the timeout
     * processing to start, it is possible that the application completes this
     * generation of async processing and starts a new one. If the timeout is
     * then processed against the new generation, response mix-up can occur.
     * This field is used to ensure that any timeout event processed is for the
     * current async generation. This prevents the response mix-up.
     */
    private volatile long asyncTimeoutGeneration = 0;//异步处理阶段
    protected final AbstractEndpoint<?> endpoint;//当前所属端点类对象。
    protected final Request request;//封装客户端请求。
    protected final Response response;//封装客户端响应。
    protected volatile SocketWrapperBase<?> socketWrapper = null;//当前socket包装类。
    protected volatile SSLSupport sslSupport;//SSL安全套接字支持。


    /**
     * Error state for the request/response currently being processed.
     */
    private ErrorState errorState = ErrorState.NONE;//处理错误状态。

    protected final UserDataHelper userDataHelper;//与日志系统结合输出无效数据

    public AbstractProcessor(AbstractEndpoint<?> endpoint) {//构造器中创建代表用户请求的Request对象和响应的Response对象。
        this(endpoint, new Request(), new Response());
    }


    protected AbstractProcessor(AbstractEndpoint<?> endpoint, Request coyoteRequest,
            Response coyoteResponse) {//完整构造器中初始化成员变量。
        this.endpoint = endpoint;
        asyncStateMachine = new AsyncStateMachine(this);
        request = coyoteRequest;
        response = coyoteResponse;
        //设置request和response的动作钩子为当前类。
        response.setHook(this);
        request.setResponse(response);
        request.setHook(this);
        userDataHelper = new UserDataHelper(getLog());
    }


    /**
     * Update the current error state to the new error state if the new error
     * state is more severe than the current error state.
     * @param errorState The error status details
     * @param t The error which occurred
     */
    protected void setErrorState(ErrorState errorState, Throwable t) {
        if (getLog().isDebugEnabled()) {
            getLog().debug(sm.getString("abstractProcessor.setErrorState", errorState), t);
        }
        // Use the return value to avoid processing more than one async error
        // in a single async cycle.
        boolean setError = response.setError();//客户端响应对象是否已经设置发生异常。
        boolean blockIo = this.errorState.isIoAllowed() && !errorState.isIoAllowed();//根据旧的错误状态和当前新的错误状态，判断是否是错误状态导致了IO阻塞。
        this.errorState = this.errorState.getMostSevere(errorState);//比较旧异常和新异常设置的级别，只保留级别更高，即最严重的异常状态。
        // Don't change the status code for IOException since that is almost
        // certainly a client disconnect in which case it is preferable to keep
        // the original status code http://markmail.org/message/4cxpwmxhtgnrwh7n
        //如果之前的响应状态小于400，并且当前设置的异常对象不是IO异常，则设置响应状态码为500。如果不是IO异常，客户端还保持连接。
        if (response.getStatus() < 400 && !(t instanceof IOException)) {
            response.setStatus(500);
        }
        //存在异常对象时，则设置request请求对象的属性值ERROR_EXCEPTION为当前异常对象t。
        if (t != null) {
            request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);
        }
        //如果异常导致IO阻塞且当前处理器为异步处理模式，同时设置异常成功，即第一次发生异常，则将状态机转换为异常状态，判断返回当前线程是否为容器线程，如果不是，将SocketEvent.ERROR错误事件放入容器线程中执行。
        if (blockIo && isAsync() && setError) {
            if (asyncStateMachine.asyncError()) {
                processSocketEvent(SocketEvent.ERROR, true);
            }
        }
    }


    protected ErrorState getErrorState() {
        return errorState;
    }


    @Override
    public Request getRequest() {
        return request;
    }


    /**
     * Set the associated adapter.
     *
     * @param adapter the new adapter
     */
    public void setAdapter(Adapter adapter) {
        this.adapter = adapter;
    }


    /**
     * Get the associated adapter.
     *
     * @return the associated adapter
     */
    public Adapter getAdapter() {
        return adapter;
    }


    /**
     * Set the socket wrapper being used.
     * @param socketWrapper The socket wrapper
     */
    protected void setSocketWrapper(SocketWrapperBase<?> socketWrapper) {
        this.socketWrapper = socketWrapper;
    }


    /**
     * @return the socket wrapper being used.
     */
    protected final SocketWrapperBase<?> getSocketWrapper() {
        return socketWrapper;
    }


    @Override
    public final void setSslSupport(SSLSupport sslSupport) {
        this.sslSupport = sslSupport;
    }


    /**
     * @return the Executor used by the underlying endpoint.
     */
    protected Executor getExecutor() {
        return endpoint.getExecutor();
    }


    @Override
    public boolean isAsync() {
        return asyncStateMachine.isAsync();
    }


    @Override
    public SocketState asyncPostProcess() {
        return asyncStateMachine.asyncPostProcess();
    }


    @Override
    public final SocketState dispatch(SocketEvent status) throws IOException {

        if (status == SocketEvent.OPEN_WRITE && response.getWriteListener() != null) {//分派执行时，客户端事件为OPEN_WRITE写事件，同时response响应对象设置的异步IO写事件不为空，此时执行写事件。
            asyncStateMachine.asyncOperation();//前面我们在状态机时看到过，该方法将状态机状态转换为READ_WRITE_OP状态。
            try {
                if (flushBufferedWrite()) {//如果刷新写缓冲区成功，则返回LONG状态，该方法由子类实现。
                    return SocketState.LONG;
                }
            } catch (IOException ioe) {//保存异常信息。
                if (getLog().isDebugEnabled()) {
                    getLog().debug("Unable to write async data.", ioe);
                }
                status = SocketEvent.ERROR;
                request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, ioe);
            }
        } else if (status == SocketEvent.OPEN_READ && request.getReadListener() != null) {//分派执行时，客户端时间为OPEN_READ读事件，同时request请求对象设置的异步IO读事件不为空，此时执行非阻塞读操作，该方法默认实现只是将状态机转为READ_WRITE_OP状态。
            dispatchNonBlockingRead();
        } else if (status == SocketEvent.ERROR) {//分派时，状态ERROR事件，需要判断之间是否设置过异常，如果没有设置过异常，则将异常写入Request请求对象的Attribute中。
            // An I/O error occurred on a non-container thread. This includes:
            // - read/write timeouts fired by the Poller (NIO & APR)
            // - completion handler failures in NIO2

            if (request.getAttribute(RequestDispatcher.ERROR_EXCEPTION) == null) {
                // Because the error did not occur on a container thread the
                // request's error attribute has not been set. If an exception
                // is available from the socketWrapper, use it to set the
                // request's error attribute here so it is visible to the error
                // handling.
                request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, socketWrapper.getError());
            }

            //读写监听器任何一个不为空，则将状态机转为READ_WRITE_OP状态。
            if (request.getReadListener() != null || response.getWriteListener() != null) {
                // The error occurred during non-blocking I/O. Set the correct
                // state else the error handling will trigger an ISE.
                asyncStateMachine.asyncOperation();
            }
        }

        RequestInfo rp = request.getRequestProcessor();//获取用户请求信息对象。
        try {
            rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);//设置当前分派步骤为STAGE_SERVICE。
            //调用适配器对象执行异步分派，如果分派失败，则设置错误状态为CLOSE_NOW，表明需要立即关闭客户端连接。
            if (!getAdapter().asyncDispatch(request, response, status)) {
                setErrorState(ErrorState.CLOSE_NOW, null);
            }
        } catch (InterruptedIOException e) {
            setErrorState(ErrorState.CLOSE_CONNECTION_NOW, e);
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            setErrorState(ErrorState.CLOSE_NOW, t);
            getLog().error(sm.getString("http11processor.request.process"), t);
        }

        rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);//设置当前分派步骤完成。

        //获取分派执行后的客户端状态。
        SocketState state;

        if (getErrorState().isError()) {//错误状态存在，则返回CLOSED客户端关闭状态。
            request.updateCounters();
            state = SocketState.CLOSED;
        } else if (isAsync()) {//异步执行，则返回LONG，表示继续持有客户端连接。
            state = SocketState.LONG;
        } else {//否则，当前请求分派完成，由子类实现dispatchEndRequest()方法决定socket的状态。
            request.updateCounters();
            state = dispatchEndRequest();
        }

        if (getLog().isDebugEnabled()) {
            getLog().debug("Socket: [" + socketWrapper +
                    "], Status in: [" + status +
                    "], State out: [" + state + "]");
        }

        return state;
    }


    protected void parseHost(MessageBytes valueMB) {
        if (valueMB == null || valueMB.isNull()) {
            populateHost();
            populatePort();
            return;
        } else if (valueMB.getLength() == 0) {
            // Empty Host header so set sever name to empty string
            request.serverName().setString("");
            populatePort();
            return;
        }

        ByteChunk valueBC = valueMB.getByteChunk();
        byte[] valueB = valueBC.getBytes();
        int valueL = valueBC.getLength();
        int valueS = valueBC.getStart();
        if (hostNameC.length < valueL) {
            hostNameC = new char[valueL];
        }

        try {
            // Validates the host name
            int colonPos = Host.parse(valueMB);

            // Extract the port information first, if any
            if (colonPos != -1) {
                int port = 0;
                for (int i = colonPos + 1; i < valueL; i++) {
                    char c = (char) valueB[i + valueS];
                    if (c < '0' || c > '9') {
                        response.setStatus(400);
                        setErrorState(ErrorState.CLOSE_CLEAN, null);
                        return;
                    }
                    port = port * 10 + c - '0';
                }
                request.setServerPort(port);

                // Only need to copy the host name up to the :
                valueL = colonPos;
            }

            // Extract the host name
            for (int i = 0; i < valueL; i++) {
                hostNameC[i] = (char) valueB[i + valueS];
            }
            request.serverName().setChars(hostNameC, 0, valueL);

        } catch (IllegalArgumentException e) {
            // IllegalArgumentException indicates that the host name is invalid
            UserDataHelper.Mode logMode = userDataHelper.getNextMode();
            if (logMode != null) {
                String message = sm.getString("abstractProcessor.hostInvalid", valueMB.toString());
                switch (logMode) {
                    case INFO_THEN_DEBUG:
                        message += sm.getString("abstractProcessor.fallToDebug");
                        //$FALL-THROUGH$
                    case INFO:
                        getLog().info(message, e);
                        break;
                    case DEBUG:
                        getLog().debug(message, e);
                }
            }

            response.setStatus(400);
            setErrorState(ErrorState.CLOSE_CLEAN, e);
        }
    }


    /**
     * Called when a host header is not present in the request (e.g. HTTP/1.0).
     * It populates the server name with appropriate information. The source is
     * expected to vary by protocol.
     * <p>
     * The default implementation is a NO-OP.
     */
    protected void populateHost() {
        // NO-OP
    }


    /**
     * Called when a host header is not present or is empty in the request (e.g.
     * HTTP/1.0). It populates the server port with appropriate information. The
     * source is expected to vary by protocol.
     * <p>
     * The default implementation is a NO-OP.
     */
    protected void populatePort() {
        // NO-OP
    }


    @Override
    public final void action(ActionCode actionCode, Object param) {
        switch (actionCode) {
            // 'Normal' servlet support
            //正常Servlet处理操作码。
            case COMMIT: {//COMMIT操作码表明提交Response时，即将发送响应客户端，此时调用prepareResponse()方法验证并构建响应头部，并且将头部数据写入缓冲区。
                if (!response.isCommitted()) {
                    try {
                        // Validate and write response headers
                        prepareResponse();
                    } catch (IOException e) {
                        handleIOException(e);
                    }
                }
                break;
            }
            case CLOSE: {//CLOSE操作码用于关闭客户端，此时首先执行COMMIT写入头部信息，随后将缓冲区中的数据写入socket。
                action(ActionCode.COMMIT, null);
                try {
                    finishResponse();
                } catch (IOException e) {
                    handleIOException(e);
                }
                break;
            }
            case ACK: {//ACK操作码用于响应HTTP1.1的状态码为100的客户端请求，ack()方法用于回复客户端是否可以继续进行请求。
                ack((ContinueResponseTiming) param);
                break;
            }
            case CLIENT_FLUSH: {//CLIENT_FLUSH操作码表明需要将缓冲区数据flush发送给客户端，要先执行COMMIT操作，构建响应头部信息。
                action(ActionCode.COMMIT, null);
                try {
                    flush();
                } catch (IOException e) {
                    handleIOException(e);
                    response.setErrorException(e);
                }
                break;
            }
            case AVAILABLE: {//AVAILABLE操作码用于获取缓冲区中可用的字节数，param用于指明是否在没有数据时进行读操作。
                request.setAvailable(available(Boolean.TRUE.equals(param)));
                break;
            }
            case REQ_SET_BODY_REPLAY: {//REQ_SET_BODY_REPLAY操作码，用于执行Realm安全校验时保存原始消息体以便重放操作。
                ByteChunk body = (ByteChunk) param;
                setRequestBody(body);
                break;
            }

            // Error handling
            //异常处理操作码。
            case IS_ERROR: {//IS_ERROR操作码用于查看当前协议处理器是否已经发生了错误信息，param参数用于保存结果。
                ((AtomicBoolean) param).set(getErrorState().isError());
                break;
            }
            case IS_IO_ALLOWED: {//IS_IO_ALLOWED操作码用于查看当前协议处理器是否执行IO操作，我们前面看到过通常是因为发生了某些异常导致处理器无法进一步执行IO操作，param参数用于保存结果。
                ((AtomicBoolean) param).set(getErrorState().isIoAllowed());
                break;
            }
            case CLOSE_NOW: {//CLOSE_NOW操作码用于在处理客户端请求时如果发生不可恢复的异常，强制关闭客户端。此时首先调用setSwallowResponse()方法写缓冲区标识为不可写状态，防止继续往缓冲区写入数据，随后根据param参数是否为异常，设置异常状态。
                // Prevent further writes to the response
                setSwallowResponse();
                if (param instanceof Throwable) {
                    setErrorState(ErrorState.CLOSE_NOW, (Throwable) param);
                } else {
                    setErrorState(ErrorState.CLOSE_NOW, null);
                }
                break;
            }
            case DISABLE_SWALLOW_INPUT: {//DISABLE_SWALLOW_INPUT操作码，调用disableSwallowRequest()方法禁止读缓冲区继续读取数据，此时有可能是禁止客户端继续大文件上传等，在后面触发hook，验证传输大小时将会执行该操作码。同时设置了CLOSE_CLEAN错误状态，表明继续使用该连接不安全，但是可以响应该请求，响应完成后应该关闭当前连接。
                // Cancelled upload or similar.
                // No point reading the remainder of the request.
                disableSwallowRequest();
                // This is an error state. Make sure it is marked as such.
                setErrorState(ErrorState.CLOSE_CLEAN, null);
                break;
            }

            // Request attribute support
            //Request Attribute域支持。
            case REQ_HOST_ADDR_ATTRIBUTE: {//REQ_HOST_ADDR_ATTRIBUTE操作码用于延迟回调提取远程主机地址。
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.remoteAddr().setString(socketWrapper.getRemoteAddr());
                }
                break;
            }
            case REQ_PEER_ADDR_ATTRIBUTE: {
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.peerAddr().setString(socketWrapper.getRemoteAddr());
                }
                break;
            }
            case REQ_HOST_ATTRIBUTE: {//REQ_PEER_ATTRIBUTE操作码用于延迟回调提取远程主机名和主机地址。
                populateRequestAttributeRemoteHost();
                break;
            }
            case REQ_LOCALPORT_ATTRIBUTE: {//REQ_LOCALPORT_ATTRIBUTE操作码用于延迟回调提取本地端口。
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.setLocalPort(socketWrapper.getLocalPort());
                }
                break;
            }
            case REQ_LOCAL_ADDR_ATTRIBUTE: {//REQ_LOCAL_ADDR_ATTRIBUTE操作码用于延迟回调提取本地地址。
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.localAddr().setString(socketWrapper.getLocalAddr());
                }
                break;
            }
            case REQ_LOCAL_NAME_ATTRIBUTE: {//REQ_LOCAL_NAME_ATTRIBUTE操作码用于延迟回调提取本地主机名。
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.localName().setString(socketWrapper.getLocalName());
                }
                break;
            }
            case REQ_REMOTEPORT_ATTRIBUTE: {//REQ_REMOTEPORT_ATTRIBUTE操作码用于延迟回调提取远程端口。
                if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
                    request.setRemotePort(socketWrapper.getRemotePort());
                }
                break;
            }

            // SSL request attribute support
            case REQ_SSL_ATTRIBUTE: {
                populateSslRequestAttributes();
                break;
            }
            case REQ_SSL_CERTIFICATE: {
                try {
                    sslReHandShake();
                } catch (IOException ioe) {
                    setErrorState(ErrorState.CLOSE_CONNECTION_NOW, ioe);
                }
                break;
            }

            // Servlet 3.0 asynchronous support
            //Servlet3.0异步支持。
            case ASYNC_START: {//ASYNC_START操作码用于异步开始时回调。
                asyncStateMachine.asyncStart((AsyncContextCallback) param);
                break;
            }
            case ASYNC_COMPLETE: {//ASYNC_COMPLETE操作码用于异步完成时回调。
                clearDispatches();
                if (asyncStateMachine.asyncComplete()) {
                    processSocketEvent(SocketEvent.OPEN_READ, true);
                }
                break;
            }
            case ASYNC_DISPATCH: {//ASYNC_DISPATCH操作码用于异步分派执行时回调。
                if (asyncStateMachine.asyncDispatch()) {
                    processSocketEvent(SocketEvent.OPEN_READ, true);
                }
                break;
            }
            case ASYNC_DISPATCHED: {//ASYNC_DISPATCH操作码用于异步分派完成时回调。
                asyncStateMachine.asyncDispatched();
                break;
            }
            case ASYNC_ERROR: {//ASYNC_ERROR操作码用于异步出错时回调。
                asyncStateMachine.asyncError();
                break;
            }
            case ASYNC_IS_ASYNC: {//ASYNC_IS_ASYNC操作码用于判断当前是否是异步执行模式。
                ((AtomicBoolean) param).set(asyncStateMachine.isAsync());
                break;
            }
            case ASYNC_IS_COMPLETING: {//ASYNC_IS_COMPLETING操作码用于判断当前是否是异步完成状态。
                ((AtomicBoolean) param).set(asyncStateMachine.isCompleting());
                break;
            }
            case ASYNC_IS_DISPATCHING: {//ASYNC_IS_DISPATCHING操作码用于判断当前是否是处于异步分派执行中。
                ((AtomicBoolean) param).set(asyncStateMachine.isAsyncDispatching());
                break;
            }
            case ASYNC_IS_ERROR: {//ASYNC_IS_ERROR操作码用于判断当前是否是异步执行且发生了错误。
                ((AtomicBoolean) param).set(asyncStateMachine.isAsyncError());
                break;
            }
            case ASYNC_IS_STARTED: {//ASYNC_IS_STARTED操作码用于判断当前异步执行是否已经开始。
                ((AtomicBoolean) param).set(asyncStateMachine.isAsyncStarted());
                break;
            }
            case ASYNC_IS_TIMINGOUT: {//ASYNC_IS_TIMINGOUT操作码用于判断当前异步执行是否超时。
                ((AtomicBoolean) param).set(asyncStateMachine.isAsyncTimingOut());
                break;
            }
            case ASYNC_RUN: {//ASYNC_RUN操作码用于异步执行传入的Runnable对象回调。
                asyncStateMachine.asyncRun((Runnable) param);
                break;
            }
            case ASYNC_SETTIMEOUT: {//ASYNC_SETTIMEOUT操作码用于设置异步执行超时时间。
                if (param == null) {
                    return;
                }
                long timeout = ((Long) param).longValue();
                setAsyncTimeout(timeout);
                break;
            }
            case ASYNC_TIMEOUT: {//ASYNC_TIMEOUT操作码用于判断当前处理器是否允许超时。
                AtomicBoolean result = (AtomicBoolean) param;
                result.set(asyncStateMachine.asyncTimeout());
                break;
            }
            case ASYNC_POST_PROCESS: {//ASYNC_POST_PROCESS操作码用于在处理错误期间触发必要的方法。
                asyncStateMachine.asyncPostProcess();
                break;
            }

            // Servlet 3.1 non-blocking I/O
            //Servlet3.1非阻塞I/O支持。
            case REQUEST_BODY_FULLY_READ: {//REQUEST_BODY_FULLY_READ操作码用于判断当前客户端发送的请求体数据是否已经读取完毕。
                AtomicBoolean result = (AtomicBoolean) param;
                result.set(isRequestBodyFullyRead());
                break;
            }
            case NB_READ_INTEREST: {//NB_READ_INTEREST操作码用于判断当前Servlet是否在接收到客户端数据时会收到通知。
                AtomicBoolean isReady = (AtomicBoolean) param;
                isReady.set(isReadyForRead());
                break;
            }
            case NB_WRITE_INTEREST: {//NB_WRITE_INTEREST操作码用于判断当前处理器是否准备好写入响应数据。
                AtomicBoolean isReady = (AtomicBoolean) param;
                isReady.set(isReadyForWrite());
                break;
            }
            case DISPATCH_READ: {//DISPATCH_READ操作码用于在数据可用读取时触发注册的异步读监听器回调。
                addDispatch(DispatchType.NON_BLOCKING_READ);
                break;
            }
            case DISPATCH_WRITE: {//DISPATCH_WRITE状态码用于在数据可写时触发注册的异步写监听器回调。
                addDispatch(DispatchType.NON_BLOCKING_WRITE);
                break;
            }
            case DISPATCH_EXECUTE: {//DISPATCH_EXECUTE操作码用于执行DISPATCH_READ、DISPATCH_WRITE回调操作。
                executeDispatches();
                break;
            }

            // Servlet 3.1 HTTP Upgrade
            case UPGRADE: {
                doHttpUpgrade((UpgradeToken) param);
                break;
            }

            // Servlet 4.0 Push requests
            case IS_PUSH_SUPPORTED: {
                AtomicBoolean result = (AtomicBoolean) param;
                result.set(isPushSupported());
                break;
            }
            case PUSH_REQUEST: {
                doPush((Request) param);
                break;
            }

            // Identifiers associated with multiplexing protocols like HTTP/2
            case CONNECTION_ID: {
                @SuppressWarnings("unchecked")
                AtomicReference<Object> result = (AtomicReference<Object>) param;
                result.set(getConnectionID());
                break;
            }
            case STREAM_ID: {
                @SuppressWarnings("unchecked")
                AtomicReference<Object> result = (AtomicReference<Object>) param;
                result.set(getStreamID());
                break;
            }
        }
    }


    private void handleIOException (IOException ioe) {
        if (ioe instanceof CloseNowException) {
            // Close the channel but keep the connection open
            setErrorState(ErrorState.CLOSE_NOW, ioe);
        } else {
            // Close the connection and all channels within that connection
            setErrorState(ErrorState.CLOSE_CONNECTION_NOW, ioe);
        }
    }


    /**
     * Perform any necessary processing for a non-blocking read before
     * dispatching to the adapter.
     */
    protected void dispatchNonBlockingRead() {
        asyncStateMachine.asyncOperation();
    }


    /**
     * {@inheritDoc}
     * <p>
     * Sub-classes of this base class represent a single request/response pair.
     * The timeout to be processed is, therefore, the Servlet asynchronous
     * processing timeout.
     */
    @Override
    public void timeoutAsync(long now) {
        if (now < 0) {//如果传入now为0，则直接执行异步超时处理，由AsyncTimeout对象的stop方法可知，当停止异步超时检测线程时，会回调方法，并传入-1。
            doTimeoutAsync();
        } else {//否则先检测是否超时，如果确定超时，则调用doTimeoutAsync()处理超时。
            long asyncTimeout = getAsyncTimeout();
            if (asyncTimeout > 0) {
                long asyncStart = asyncStateMachine.getLastAsyncStart();
                if ((now - asyncStart) > asyncTimeout) {//当前时间-异步开始时间判断超时。
                    doTimeoutAsync();
                }
            } else if (!asyncStateMachine.isAvailable()) {//进一步判断当前状态机是否有效，如果无效则处理超时。
                // Timeout the async process if the associated web application
                // is no longer running.
                doTimeoutAsync();
            }
        }
    }


    private void doTimeoutAsync() {
        // Avoid multiple timeouts
        setAsyncTimeout(-1);//首先，将超时时间设置为-1，表明不执行超时，避免触发多次超时动作。
        asyncTimeoutGeneration = asyncStateMachine.getCurrentGeneration();//保存当前处理超时的周期。
        processSocketEvent(SocketEvent.TIMEOUT, true);//执行TIMEOUT超时事件。
    }


    @Override
    public boolean checkAsyncTimeoutGeneration() {
        return asyncTimeoutGeneration == asyncStateMachine.getCurrentGeneration();
    }


    public void setAsyncTimeout(long timeout) {
        asyncTimeout = timeout;
    }


    public long getAsyncTimeout() {
        return asyncTimeout;
    }


    @Override
    public void recycle() {
        errorState = ErrorState.NONE;
        asyncStateMachine.recycle();
    }


    protected abstract void prepareResponse() throws IOException;//由子类负责实现。


    protected abstract void finishResponse() throws IOException;//由子类负责实现。


    /**
     * @deprecated Unused. This will be removed in Tomcat 10 onwards. Use
     *             {@link #ack(ContinueResponseTiming)}.
     */
    @Deprecated
    protected void ack() {
        ack(ContinueResponseTiming.ALWAYS);
    }


    protected abstract void ack(ContinueResponseTiming continueResponseTiming);//由子类负责实现。


    protected abstract void flush() throws IOException;//由子类负责实现。


    protected abstract int available(boolean doRead);//由子类负责实现。


    protected abstract void setRequestBody(ByteChunk body);//由子类负责实现。


    protected abstract void setSwallowResponse();//由子类负责实现。


    protected abstract void disableSwallowRequest();//由子类负责实现。


    /**
     * Processors that populate request attributes directly (e.g. AJP) should
     * over-ride this method and return {@code false}.
     *
     * @return {@code true} if the SocketWrapper should be used to populate the
     *         request attributes, otherwise {@code false}.
     */
    protected boolean getPopulateRequestAttributesFromSocket() {
        return true;
    }


    /**
     * Populate the remote host request attribute. Processors (e.g. AJP) that
     * populate this from an alternative source should override this method.
     */
    protected void populateRequestAttributeRemoteHost() {
        if (getPopulateRequestAttributesFromSocket() && socketWrapper != null) {
            request.remoteHost().setString(socketWrapper.getRemoteHost());
        }
    }


    /**
     * Populate the TLS related request attributes from the {@link SSLSupport}
     * instance associated with this processor. Protocols that populate TLS
     * attributes from a different source (e.g. AJP) should override this
     * method.
     */
    protected void populateSslRequestAttributes() {
        try {
            if (sslSupport != null) {
                Object sslO = sslSupport.getCipherSuite();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.CIPHER_SUITE_KEY, sslO);
                }
                sslO = sslSupport.getPeerCertificateChain();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.CERTIFICATE_KEY, sslO);
                }
                sslO = sslSupport.getKeySize();
                if (sslO != null) {
                    request.setAttribute (SSLSupport.KEY_SIZE_KEY, sslO);
                }
                sslO = sslSupport.getSessionId();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.SESSION_ID_KEY, sslO);
                }
                sslO = sslSupport.getProtocol();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.PROTOCOL_VERSION_KEY, sslO);
                }
                sslO = sslSupport.getRequestedProtocols();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.REQUESTED_PROTOCOL_VERSIONS_KEY, sslO);
                }
                sslO = sslSupport.getRequestedCiphers();
                if (sslO != null) {
                    request.setAttribute(SSLSupport.REQUESTED_CIPHERS_KEY, sslO);
                }
                request.setAttribute(SSLSupport.SESSION_MGR, sslSupport);
            }
        } catch (Exception e) {
            getLog().warn(sm.getString("abstractProcessor.socket.ssl"), e);
        }
    }


    /**
     * Processors that can perform a TLS re-handshake (e.g. HTTP/1.1) should
     * override this method and implement the re-handshake.
     *
     * @throws IOException If authentication is required then there will be I/O
     *                     with the client and this exception will be thrown if
     *                     that goes wrong
     */
    protected void sslReHandShake() throws IOException {
        // NO-OP
    }


    protected void processSocketEvent(SocketEvent event, boolean dispatch) {
        SocketWrapperBase<?> socketWrapper = getSocketWrapper();
        if (socketWrapper != null) {
            socketWrapper.processSocket(event, dispatch);
        }
    }


    protected boolean isReadyForRead() {
        if (available(true) > 0) {
            return true;
        }

        if (!isRequestBodyFullyRead()) {
            registerReadInterest();
        }

        return false;
    }


    protected abstract boolean isRequestBodyFullyRead();


    protected abstract void registerReadInterest();


    protected abstract boolean isReadyForWrite();


    protected void executeDispatches() {
        SocketWrapperBase<?> socketWrapper = getSocketWrapper();
        Iterator<DispatchType> dispatches = getIteratorAndClearDispatches();
        if (socketWrapper != null) {
            synchronized (socketWrapper) {
                /*
                 * This method is called when non-blocking IO is initiated by defining
                 * a read and/or write listener in a non-container thread. It is called
                 * once the non-container thread completes so that the first calls to
                 * onWritePossible() and/or onDataAvailable() as appropriate are made by
                 * the container.
                 *
                 * Processing the dispatches requires (for APR/native at least)
                 * that the socket has been added to the waitingRequests queue. This may
                 * not have occurred by the time that the non-container thread completes
                 * triggering the call to this method. Therefore, the coded syncs on the
                 * SocketWrapper as the container thread that initiated this
                 * non-container thread holds a lock on the SocketWrapper. The container
                 * thread will add the socket to the waitingRequests queue before
                 * releasing the lock on the socketWrapper. Therefore, by obtaining the
                 * lock on socketWrapper before processing the dispatches, we can be
                 * sure that the socket has been added to the waitingRequests queue.
                 */
                //遍历所有类型，并调用socketWrapper.processSocket()执行DispatchType中指定的SocketEvent。
                while (dispatches != null && dispatches.hasNext()) {
                    DispatchType dispatchType = dispatches.next();
                    socketWrapper.processSocket(dispatchType.getSocketStatus(), false);
                }
            }
        }
    }


    /**
     * {@inheritDoc}
     * Processors that implement HTTP upgrade must override this method and
     * provide the necessary token.
     */
    @Override
    public UpgradeToken getUpgradeToken() {
        // Should never reach this code but in case we do...
        throw new IllegalStateException(
                sm.getString("abstractProcessor.httpupgrade.notsupported"));
    }


    /**
     * Process an HTTP upgrade. Processors that support HTTP upgrade should
     * override this method and process the provided token.
     *
     * @param upgradeToken Contains all the information necessary for the
     *                     Processor to process the upgrade
     *
     * @throws UnsupportedOperationException if the protocol does not support
     *         HTTP upgrade
     */
    protected void doHttpUpgrade(UpgradeToken upgradeToken) {
        // Should never happen
        throw new UnsupportedOperationException(
                sm.getString("abstractProcessor.httpupgrade.notsupported"));
    }


    /**
     * {@inheritDoc}
     * Processors that implement HTTP upgrade must override this method.
     */
    @Override
    public ByteBuffer getLeftoverInput() {
        // Should never reach this code but in case we do...
        throw new IllegalStateException(sm.getString("abstractProcessor.httpupgrade.notsupported"));
    }


    /**
     * {@inheritDoc}
     * Processors that implement HTTP upgrade must override this method.
     */
    @Override
    public boolean isUpgrade() {
        return false;
    }


    /**
     * Protocols that support push should override this method and return {@code
     * true}.
     *
     * @return {@code true} if push is supported by this processor, otherwise
     *         {@code false}.
     */
    protected boolean isPushSupported() {
        return false;
    }


    /**
     * Process a push. Processors that support push should override this method
     * and process the provided token.
     *
     * @param pushTarget Contains all the information necessary for the Processor
     *                   to process the push request
     *
     * @throws UnsupportedOperationException if the protocol does not support
     *         push
     */
    protected void doPush(Request pushTarget) {
        throw new UnsupportedOperationException(
                sm.getString("abstractProcessor.pushrequest.notsupported"));
    }


    /**
     * Protocols that support multiplexing (e.g. HTTP/2) should override this
     * method and return the appropriate ID.
     *
     * @return The stream ID associated with this request or {@code null} if a
     *         multiplexing protocol is not being used
      */
    protected Object getConnectionID() {
        return null;
    }


    /**
     * Protocols that support multiplexing (e.g. HTTP/2) should override this
     * method and return the appropriate ID.
     *
     * @return The stream ID associated with this request or {@code null} if a
     *         multiplexing protocol is not being used
     */
    protected Object getStreamID() {
        return null;
    }


    /**
     * Flush any pending writes. Used during non-blocking writes to flush any
     * remaining data from a previous incomplete write.
     *
     * @return <code>true</code> if data remains to be flushed at the end of
     *         method
     *
     * @throws IOException If an I/O error occurs while attempting to flush the
     *         data
     */
    protected abstract boolean flushBufferedWrite() throws IOException ;


    /**
     * Perform any necessary clean-up processing if the dispatch resulted in the
     * completion of processing for the current request.
     *
     * @return The state to return for the socket once the clean-up for the
     *         current request has completed
     *
     * @throws IOException If an I/O error occurs while attempting to end the
     *         request
     */
    protected abstract SocketState dispatchEndRequest() throws IOException;


    @Override
    protected final void logAccess(SocketWrapperBase<?> socketWrapper) throws IOException {
        // Set the socket wrapper so the access log can read the socket related
        // information (e.g. client IP)
        setSocketWrapper(socketWrapper);
        // Setup the minimal request information
        request.setStartTime(System.currentTimeMillis());
        // Setup the minimal response information
        response.setStatus(400);
        response.setError();
        getAdapter().log(request, response, 0);
    }
}
